package util

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/go-redis/redis/v9"
	"gorm.io/gorm"
	"likeadmin/core"
	"likeadmin/core/response"
	"likeadmin/model/client"
)

// 常量定义
const (
	OrderQueueKey      = "client:backstage:order:queue"
	OrderProcessedKey  = "order_processed:%s"
	ProcessInterval    = 1 * time.Second
	ProcessMarkTTL     = time.Hour
	EventTypeActive    = "active"
	DefaultEventType   = "recharge"
	OceanEngineChannel = 1
)

// OrderMessage 订单消息结构体
type OrderMessage struct {
	Type     string    `json:"type"`      // 事件类型
	OrderID  string    `json:"order_id"`  // 订单唯一标识符
	DeviceId string    `json:"device_id"` // 用户唯一标识符
	Complete bool      `json:"complete"`  // 标识订单是否已完成
	Time     time.Time `json:"time"`      // 消息生成的时间
}

// OrderProcessor 订单处理器接口
type OrderProcessor interface {
	Process(msg OrderMessage) error
}

// QueueService 队列服务结构体
type QueueService struct {
	rdb        *redis.Client
	ctx        context.Context
	cancel     context.CancelFunc
	processors map[string]OrderProcessor
}

// NewQueueService 创建新的队列服务实例
func NewQueueService(redisClient *redis.Client) *QueueService {
	ctx, cancel := context.WithCancel(context.Background())
	qs := &QueueService{
		rdb:        redisClient,
		ctx:        ctx,
		cancel:     cancel,
		processors: make(map[string]OrderProcessor),
	}

	qs.registerProcessors()
	go qs.orderQueueWorker()

	core.Logger.Info("队列服务初始化完成")
	return qs
}

// registerProcessors 注册消息处理器
func (qs *QueueService) registerProcessors() {
	qs.processors[EventTypeActive] = &ActiveOrderProcessor{}
	qs.processors[DefaultEventType] = &RechargeOrderProcessor{queueService: qs}
}

// Close 关闭队列服务
func (qs *QueueService) Close() {
	qs.cancel()
	core.Logger.Info("队列服务已关闭")
}

// AddOrderMessage 添加订单消息到延迟队列
func (qs *QueueService) AddOrderMessage(msg OrderMessage, delaySeconds int64) error {
	core.Logger.Debugf("添加订单消息到队列: OrderID=%s, Type=%s", msg.OrderID, msg.Type)

	msgJson, err := json.Marshal(msg)
	if err != nil {
		return fmt.Errorf("序列化订单消息失败: %w", err)
	}

	score := time.Now().Unix() + delaySeconds
	return qs.rdb.ZAdd(qs.ctx, OrderQueueKey, redis.Z{
		Score:  float64(score),
		Member: msgJson,
	}).Err()
}

// orderQueueWorker 订单队列处理工作协程
func (qs *QueueService) orderQueueWorker() {
	core.Logger.Info("启动订单队列处理工作协程")
	defer core.Logger.Info("订单队列处理工作协程结束")

	ticker := time.NewTicker(ProcessInterval)
	defer ticker.Stop()

	for {
		select {
		case <-ticker.C:
			qs.processDueMessages()
		case <-qs.ctx.Done():
			return
		}
	}
}

// processDueMessages 处理到期的消息
func (qs *QueueService) processDueMessages() {
	now := time.Now().Unix()
	results, err := qs.rdb.ZRangeByScore(qs.ctx, OrderQueueKey, &redis.ZRangeBy{
		Min: "0",
		Max: fmt.Sprintf("%d", now),
	}).Result()

	if err != nil {
		core.Logger.Errorf("获取到期消息失败: %v", err)
		return
	}

	if len(results) == 0 {
		return
	}

	qs.batchProcessMessages(results)
}

// batchProcessMessages 批量处理消息
func (qs *QueueService) batchProcessMessages(messages []string) {
	pipe := qs.rdb.TxPipeline()
	processedCount := 0
	failedCount := 0

	for _, msgJson := range messages {
		var msg OrderMessage
		if err := json.Unmarshal([]byte(msgJson), &msg); err != nil {
			core.Logger.Errorf("反序列化消息失败: %v", err)
			failedCount++
			continue
		}

		if qs.processSingleMessage(msg) {
			pipe.ZRem(qs.ctx, OrderQueueKey, msgJson)
			processedCount++
		} else {
			failedCount++
		}
	}

	if processedCount > 0 {
		if _, err := pipe.Exec(qs.ctx); err != nil {
			core.Logger.Errorf("删除已处理消息失败: %v", err)
		} else {
			core.Logger.Debugf("消息处理完成: 成功=%d, 失败=%d", processedCount, failedCount)
		}
	}
}

// processSingleMessage 处理单个消息
func (qs *QueueService) processSingleMessage(msg OrderMessage) bool {
	core.Logger.Infof("开始处理订单消息: OrderID=%s, Type=%s", msg.OrderID, msg.Type)

	// 幂等性检查
	if !qs.acquireProcessLock(msg.OrderID) {
		core.Logger.Infof("订单消息已处理过，跳过: OrderID=%s", msg.OrderID)
		return true
	}

	defer qs.releaseProcessLock(msg.OrderID)

	// 根据消息类型选择处理器
	processor, exists := qs.processors[msg.Type]
	if !exists {
		// 使用默认处理器
		processor = qs.processors[DefaultEventType]
	}

	if err := processor.Process(msg); err != nil {
		core.Logger.Errorf("处理订单消息失败: OrderID=%s, Error=%v", msg.OrderID, err)
		return false
	}

	core.Logger.Infof("订单消息处理成功: OrderID=%s", msg.OrderID)
	return true
}

// acquireProcessLock 获取处理锁（幂等性检查）
func (qs *QueueService) acquireProcessLock(orderID string) bool {
	key := fmt.Sprintf(OrderProcessedKey, orderID)
	result := qs.rdb.SetNX(qs.ctx, key, "1", ProcessMarkTTL)
	if result.Err() != nil {
		core.Logger.Errorf("设置处理标记失败: OrderID=%s, Error=%v", orderID, result.Err())
		return false
	}
	return result.Val()
}

// releaseProcessLock 释放处理锁
func (qs *QueueService) releaseProcessLock(orderID string) {
	key := fmt.Sprintf(OrderProcessedKey, orderID)
	if err := qs.rdb.Del(qs.ctx, key).Err(); err != nil {
		core.Logger.Errorf("删除处理标记失败: OrderID=%s, Error=%v", orderID, err)
	}
}

// ActiveOrderProcessor 激活事件处理器
type ActiveOrderProcessor struct{}

// Process 处理激活事件
func (p *ActiveOrderProcessor) Process(msg OrderMessage) error {
	return activeOceanengine(msg)
}

// RechargeOrderProcessor 充值订单处理器
type RechargeOrderProcessor struct {
	queueService *QueueService
}

// Process 处理充值订单
func (p *RechargeOrderProcessor) Process(msg OrderMessage) error {
	order, err := p.getPayOrder(msg.OrderID)
	if err != nil {
		return fmt.Errorf("获取支付订单失败: %w", err)
	}

	account, source, err := p.validateBusinessData(order)
	if err != nil {
		return fmt.Errorf("验证业务数据失败: %w", err)
	}

	return p.executeRechargeBusiness(order, account, source)
}

// getPayOrder 查询支付订单
func (p *RechargeOrderProcessor) getPayOrder(orderID string) (*client.Orders, error) {
	var order client.Orders
	err := core.GetDB().Model(&client.Orders{}).
		Where("out_no = ? AND is_pay = ?", orderID, 1).
		First(&order).Error

	if errors.Is(err, gorm.ErrRecordNotFound) {
		return nil, fmt.Errorf("订单未找到或未支付: OrderID=%s", orderID)
	}
	if err != nil {
		return nil, fmt.Errorf("订单查询失败: %w", err)
	}

	core.Logger.Debugf("找到订单: OrderID=%s, Amount=%d", order.OutNo, order.Money)
	return &order, nil
}

// validateBusinessData 验证业务数据
func (p *RechargeOrderProcessor) validateBusinessData(order *client.Orders) (*client.Account, *client.Source, error) {
	account, err := p.getAccount(order.AccountId, order.PackageName)
	if err != nil {
		return nil, nil, err
	}

	source, err := p.getSource(order.SourceId, order.PackageName)
	if err != nil {
		return nil, nil, err
	}

	return account, source, nil
}

// getAccount 获取广告账户
func (p *RechargeOrderProcessor) getAccount(accountID string, packageName string) (*client.Account, error) {
	var account client.Account
	err := core.GetDB().Model(&client.Account{}).
		Where("account_id = ? AND package_name = ?", accountID, packageName).
		First(&account).Error

	if err != nil {
		return nil, fmt.Errorf("广告账户查询失败: AccountID=%s, Package=%s, Error=%w",
			accountID, packageName, err)
	}
	return &account, nil
}

// getSource 获取外推渠道
func (p *RechargeOrderProcessor) getSource(sourceID, packageName string) (*client.Source, error) {
	var source client.Source
	err := core.GetDB().Model(&client.Source{}).
		Where("source_id = ? AND package_name = ?", sourceID, packageName).
		First(&source).Error

	if err != nil {
		return nil, fmt.Errorf("外推渠道查询失败: SourceID=%s, Package=%s, Error=%w",
			sourceID, packageName, err)
	}
	return &source, nil
}

// executeRechargeBusiness 执行充值业务逻辑
func (p *RechargeOrderProcessor) executeRechargeBusiness(order *client.Orders, account *client.Account, source *client.Source) error {
	tx := core.GetDB().Begin()
	if tx.Error != nil {
		return fmt.Errorf("开启事务失败: %w", tx.Error)
	}

	var err error
	defer func() {
		if r := recover(); r != nil {
			tx.Rollback()
			core.Logger.Errorf("处理订单消息异常回滚: OrderID=%s, Error=%v", order.OutNo, r)
		} else if err != nil {
			tx.Rollback()
		}
	}()

	//if err = p.updateAccountRecharge(tx, account, order.Money); err != nil {
	//	return err
	//}
	//
	//if err = p.updateSourceRecharge(tx, source, order.Money); err != nil {
	//	return err
	//}
	//
	//if err = p.updateDailyRechargeReport(tx, order, source); err != nil {
	//	return err
	//}

	if err = tx.Commit().Error; err != nil {
		return fmt.Errorf("提交事务失败: %w", err)
	}

	// 根据用户的ua前面一段+ip+渠道id 匹配监测链接的数据
	go func() {
		err := p.matchCallbackInfo(order)
		if err != nil {
			core.Logger.Infof("匹配监测信息失败: OrderID=%s, Account=%s, Source=%s, Amount=%d",
				order.OutNo, account.AccountId, source.SourceId, order.Money)
		}
	}()

	core.Logger.Infof("处理订单成功: OrderID=%s, Account=%s, Source=%s, Amount=%d",
		order.OutNo, account.AccountId, source.SourceId, order.Money)
	return nil
}

// getUAPrefix 获取UA前缀
func getUAPrefix(ua string, maxLength int) string {
	ua = strings.ToLower(ua)
	if len(ua) > maxLength {
		return ua[:maxLength]
	}
	return ua
}

// matchCallbackInfo 匹配监测链接数据
func (p *RechargeOrderProcessor) matchCallbackInfo(order *client.Orders) error {
	// 截取UA前面一段进行匹配（例如前32个字符）
	uaPrefix := getUAPrefix(order.PayUa, 32)

	// 查询匹配的回调信息
	var callbackInfo client.CallbackInfo
	err := core.GetDB().Model(&client.CallbackInfo{}).
		Where("ip = ? AND source_id = ? AND ua LIKE ? AND create_time > ?", order.PayIp, order.SourceId, uaPrefix+"%", order.PayTime-10800).
		Order("id desc").
		First(&callbackInfo).Error

	if err != nil {
		core.Logger.Errorf("查询回调信息失败: OrderID=%s, Error=%v", order.OutNo, err)
		return err
	}

	// 更新订单表中的回调信息字段
	return p.updateOrderCallbackInfo(order, &callbackInfo)
}

// randomMatchUnprocessedCallbackInfo 随机匹配未回传的监测信息
func (p *RechargeOrderProcessor) randomMatchUnprocessedCallbackInfo(order *client.Orders) error {
	var callbackInfo client.CallbackInfo

	// 查询相同渠道下未回传的监测信息，随机取一条
	err := core.GetDB().Model(&client.CallbackInfo{}).
		Where("source_id = ? AND callback_status < ?", order.SourceId, 2).
		Order("RAND()"). // 随机排序
		First(&callbackInfo).Error

	if err != nil {
		if errors.Is(err, gorm.ErrRecordNotFound) {
			core.Logger.Infof("未找到任何未回传的监测信息: OrderID=%s, SourceID=%s", order.OutNo, order.SourceId)
			return nil
		}
		core.Logger.Errorf("随机查询回调信息失败: OrderID=%s, Error=%v", order.OutNo, err)
		return err
	}

	core.Logger.Infof("随机匹配到未回传监测信息: OrderID=%s, CallbackInfoID=%d", order.OutNo, callbackInfo.ID)

	// 更新订单表中的回调信息字段
	return p.updateOrderCallbackInfo(order, &callbackInfo)
}

// updateOrderCallbackInfo 更新订单回调信息
func (p *RechargeOrderProcessor) updateOrderCallbackInfo(order *client.Orders, callbackInfo *client.CallbackInfo) error {
	// 更新订单表中的回调信息字段
	updates := map[string]interface{}{
		"project_id":       callbackInfo.ProjectId,
		"project_name":     callbackInfo.ProjectName,
		"promotion_id":     callbackInfo.PromotionId,
		"promotion_name":   callbackInfo.PromotionName,
		"callback_info_id": callbackInfo.ID,
	}

	err := core.GetDB().Model(&client.Orders{}).Where("out_no = ?", order.OutNo).Select("project_id", "project_name", "promotion_id", "promotion_name", "callback_info_id").Updates(updates).Error
	if err != nil {
		core.Logger.Errorf("更新订单回调信息失败: OrderID=%s, Error=%v", order.OutNo, err)
		return err
	}

	core.Logger.Infof("订单匹配回调信息成功: OrderID=%s, CallbackInfoID=%d", order.OutNo, callbackInfo.ID)
	return nil
}

// updateAccountRecharge 更新账户充值金额
func (p *RechargeOrderProcessor) updateAccountRecharge(tx *gorm.DB, account *client.Account, money uint) error {
	return tx.Model(&client.Account{}).
		Where("account_id = ?", account.AccountId).
		Updates(map[string]interface{}{
			"recharge":    account.Recharge + money,
			"update_time": time.Now().Unix(),
		}).Error
}

// updateSourceRecharge 更新渠道充值金额
func (p *RechargeOrderProcessor) updateSourceRecharge(tx *gorm.DB, source *client.Source, money uint) error {
	return tx.Model(&client.Source{}).
		Where("source_id = ?", source.SourceId).
		Updates(map[string]interface{}{
			"recharge":    source.Recharge + money,
			"update_time": time.Now().Unix(),
		}).Error
}

// updateDailyRechargeReport 更新每日充值报表
func (p *RechargeOrderProcessor) updateDailyRechargeReport(tx *gorm.DB, order *client.Orders, source *client.Source) error {
	// 根据用户实际支付的时间，归属统计到那一天
	payDate := time.Unix(order.CreateTime, 0).Format("2006-01-02")
	payYear := time.Unix(order.CreateTime, 0).Format("2006")
	payMonth := time.Unix(order.CreateTime, 0).Format("2006-01")

	var report client.RechargeReport

	err := tx.Model(&client.RechargeReport{}).
		Where("account_id = ? AND date = ?", order.AccountId, payDate).
		First(&report).Error

	now := time.Now().Unix()
	if errors.Is(err, gorm.ErrRecordNotFound) {
		return p.createDailyRechargeReport(tx, order, source, payDate, payYear, payMonth, now)
	} else if err != nil {
		return fmt.Errorf("查询日报表失败: %w", err)
	}

	return p.updateExistingRechargeReport(tx, order, payDate, now)
}

// createDailyRechargeReport 创建新的日报表记录
func (p *RechargeOrderProcessor) createDailyRechargeReport(tx *gorm.DB, order *client.Orders, source *client.Source, date, year, month string, timestamp int64) error {
	return tx.Model(&client.RechargeReport{}).Create(map[string]interface{}{
		"source_id":      source.SourceId,
		"account_id":     source.AccountId,
		"creater_id":     source.CreaterID,
		"recharge":       order.Money,
		"total_recharge": order.Money, // 新增累计充值字段
		"year":           year,
		"month":          month,
		"date":           date,
		"create_time":    timestamp,
		"update_time":    timestamp,
	}).Error
}

// updateExistingRechargeReport 更新现有的日报表记录
func (p *RechargeOrderProcessor) updateExistingRechargeReport(tx *gorm.DB, order *client.Orders, date string, timestamp int64) error {
	return tx.Model(&client.RechargeReport{}).
		Where("account_id = ? AND date = ?", order.AccountId, date).
		Updates(map[string]interface{}{
			"recharge":       gorm.Expr("recharge + ?", order.Money),
			"total_recharge": gorm.Expr("total_recharge + ?", order.Money),
			"orders":         gorm.Expr("orders + 1"),
			"update_time":    timestamp,
		}).Error
}

// 头条激活处理
func activeOceanengine(msg OrderMessage) error {
	clientUser, err := getClientUserByDevice(msg.DeviceId)
	if err != nil {
		return fmt.Errorf("查询用户设备失败: %w", err)
	}

	callbackInfo, err := getCallbackInfo(clientUser.SourceId, clientUser.RegisterIp)
	if err != nil {
		return fmt.Errorf("查询回调信息失败: %w", err)
	}

	if callbackInfo.Channel != OceanEngineChannel {
		return fmt.Errorf("监测链接类型不是头条: Channel=%d", callbackInfo.Channel)
	}

	callbackURL, err := parseCallbackURL(callbackInfo.Link)
	if err != nil {
		return fmt.Errorf("解析回调链接失败: %w", err)
	}

	if err := executeOceanEngineCallback(callbackURL, callbackInfo.CreateTime); err != nil {
		return fmt.Errorf("执行头条回调失败: %w", err)
	}

	core.Logger.Infof("头条激活成功: DeviceID=%s, CallbackURL=%s", msg.DeviceId, callbackURL)
	return nil
}

// getClientUserByDevice 根据设备ID获取用户信息
func getClientUserByDevice(deviceID string) (*client.ClientUser, error) {
	var clientUser client.ClientUser
	err := core.GetDB().Model(&client.ClientUser{}).
		Where("device_id = ?", deviceID).
		First(&clientUser).Error

	if response.CheckErrDBNotRecord(err, "ClientUser First err") != nil {
		return nil, fmt.Errorf("查询用户设备失败: %w", err)
	}
	return &clientUser, nil
}

// getCallbackInfo 获取回调信息
func getCallbackInfo(sourceID, ip string) (*client.CallbackInfo, error) {
	var callbackInfo client.CallbackInfo
	err := core.GetDB().Model(&client.CallbackInfo{}).
		Where("source_id = ? AND ip = ? AND callback_status = ?", sourceID, ip, 0).
		Order("id desc").
		First(&callbackInfo).Error

	if response.CheckErrDBNotRecord(err, "callbackInfo First err") != nil {
		return nil, fmt.Errorf("查询回调信息失败: %w", err)
	}
	return &callbackInfo, nil
}

// parseCallbackURL 解析回调URL
func parseCallbackURL(link string) (string, error) {
	var reportData map[string]interface{}
	if err := ToolsUtil.JsonToObj(link, &reportData); err != nil {
		return "", fmt.Errorf("JSON解析失败: %w", err)
	}

	callbackURL, ok := reportData["callback_url"].(string)
	if !ok || callbackURL == "" {
		return "", fmt.Errorf("回调链接为空或格式错误")
	}

	return callbackURL, nil
}

// executeOceanEngineCallback 执行头条回调
func executeOceanEngineCallback(callbackURL string, createTime int64) error {
	result, err := OceanengineCallbackUtil{}.Callback(&AdCallback{
		CallbackURL:     callbackURL,
		Action:          "ACTIVATION",
		MicroCreateTime: createTime * 1000, // 转换为毫秒
		Money:           0,
	})

	if err != nil {
		return fmt.Errorf("回调上报失败: %w", err)
	}

	if strings.Contains(result, "::error") {
		return fmt.Errorf("回调上报返回错误: %s", result)
	}

	return nil
}
